package com.kt.common.coroutines

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.system.measureTimeMillis

//挂起
/*
suspend fun simple(): List<Int> {
    delay(1000) // 假装我们在这里做了一些异步的事情
    return listOf(1, 2, 3)
}

fun main() = runBlocking<Unit> {
    simple().forEach { value -> println(value) }
}*/


/*fun simple(): Flow<Int> = flow {
    for (i in 1..3) {
        emit(i)
        //delay(100L)
        Thread.sleep(100)
    }
}*/

/*fun main() = runBlocking<Unit> {
    launch {
        for (k in 1..3) {
            println("I'm not blocked $k")
            delay(100)
        }
    }
    println("this runBlocking before")
    simple().collect {
        println(it)
    }
    println("this runBlocking after")
}*/


/*
fun main() = runBlocking {
    println("Calling simple function...")
    val flow = simple()
    println("Calling collect...")
    flow.collect { value -> println(value) }
    println("Calling collect again...")
    flow.collect { println(it) }
}*/

//取消流
/*fun main() = runBlocking {
    withTimeoutOrNull(150) { // 在 250 毫秒后超时
        simple().collect { value -> println(value) }
    }
    println("Done")
}*/

//创建流的方式
/*fun main() = runBlocking<Unit> {
    //通过asFlow来测试
   *//* (1..5).asFlow().collect {
        delay(100L)
        println(it)
    }*//*

    //通过flowOf来测试
    flowOf(1,3,4,5,6,7).collect {
        delay(100L)
        println(it)
    }
}*/


suspend fun performRequest(request: Int): String {
    delay(1000) // 模仿长时间运行的异步工作
    return "response $request"
}
// 过滤流操作符
/*fun main() = runBlocking<Unit> {
    (1..3).asFlow()
        .map { performRequest(it) }
        .collect {
            println(it)
        }
}*/

//转换操作符
/*
fun main() = runBlocking<Unit> {
    (1..3).asFlow()
        .transform {
            emit(performRequest(it))
            emit("Making request $it")
        }
        .collect {
            println(it)
        }
}*/

//限制过滤操作符
/*
fun main() = runBlocking<Unit> {
    flow {
        try {
            emit(1)
            emit(2)
            println("This line will not execute")
            emit(3)
        } finally {
            println("Finally in numbers")
        }

    }
        .take(2)
        .collect {
            println(it)
        }
}*/


//末端操作符
/*
fun main() = runBlocking<Unit> {
    val sum = (1..5).asFlow()
        .map { it * it }
        .reduce { accumulator, value ->
            println("accumulator = $accumulator ,value=$value")
            accumulator + value
        }
    println(sum)
}*/

//流的连续
/*
fun main() = runBlocking<Unit> {
    //filter 2
    //filter 3
    //Map 3
    //Collect string 3
    //filter 4
    //Map 4
    //Collect string 4
    (2..4).asFlow()
        .filter {
            println("filter $it")
            it>2
        }
        .map {
            println("Map $it")
            "string $it"
        }
        .collect {
            println("Collect $it")
        }
}*/

//改变上下文
/*fun simple(): Flow<Int> = flow {
    // 在流构建器中更改消耗 CPU 代码的上下文的错误方式
    withContext(Dispatchers.Default){
        for (i in 1..3) {
            Thread.sleep(100)
            emit(i)
            //delay(100L)

        }
    }

}*/

//正确改变flow 上下文的方式
// threadName = DefaultDispatcher-worker-1
//value = 1 ,threadName = main
// threadName = DefaultDispatcher-worker-1
//value = 2 ,threadName = main
// threadName = DefaultDispatcher-worker-1
//value = 3 ,threadName = main
/*
fun simple(): Flow<Int> = flow {

    for (i in 1..3) {
        Thread.sleep(100)
        emit(i)
        println(" threadName = ${Thread.currentThread().name}")
        //delay(100L)
    }
}.flowOn(Dispatchers.Default)// 在流构建器中改变消耗 CPU 代码上下文的正确方式

fun main() = runBlocking<Unit> {
    simple().collect { value -> println("value = $value ,threadName = ${Thread.currentThread().name}") }
}*/


/**缓冲*/
fun simple(): Flow<Int> = flow {
    for (i in 1..3) {
        delay(100) // 假装我们异步等待了 100 毫秒
        emit(i) // 发射下一个值
    }
}

/*
fun main() = runBlocking<Unit> {
    val time = measureTimeMillis {
        simple().collect { value ->
            delay(300) // 假装我们花费 300 毫秒来处理它
            println(value)
        }
    }
    println("Collected in $time ms")
}
*/
//它产生了相同的数字，只是更快了，由于我们高效地创建了处理流水线， 仅仅需要等待第一个数字产生的 100 毫秒以及处理每个数字各需花费的 300 毫秒
/*fun main() = runBlocking<Unit> {
    val time = measureTimeMillis {
        simple()
            .buffer()
            .collect { value ->
                delay(300) // 假装我们花费 300 毫秒来处理它
                println(value)
            }
    }
    println("Collected in $time ms")
}*/


/*
fun main() = runBlocking<Unit> {
    //1
    //3
    //Collected in 875 ms
    //从结果可以看出来使用conflate操作符可以跳过中间值
    val time = measureTimeMillis {
        simple()
            .conflate()
            .collect { value ->
                delay(300) // 假装我们花费 300 毫秒来处理它
                println(value)
            }
    }
    println("Collected in $time ms")
}*/


/*
fun main() = runBlocking<Unit> {
    //Collecting 1
    //Collecting 2
    //Collecting 3
    //Done 3
    //Collected in 781 ms
    //由于 collectLatest 的函数体需要花费 300 毫秒，但是新值每 100 秒发射一次，我们看到该代码块对每个值运行，但是只收集最后一个值：
    val time = measureTimeMillis {
        simple()
            .collectLatest{ value ->
                println("Collecting $value")
                delay(300) // 假装我们花费 300 毫秒来处理它
                println("Done $value")
            }
    }
    println("Collected in $time ms")
}*/


/*
fun main() = runBlocking<Unit> {
    val nums = (1..3).asFlow() // 数字 1..3
    val strs = flowOf("one", "two", "three") // 字符串
    nums.zip(strs) { a, b ->
        "$a -> $b"
    }
        .collect {
            println(it)
        }

}*/

//
/*
fun main() = runBlocking<Unit> {
    //1 -> one at 531 ms from start
    //2 -> two at 922 ms from start
    //3 -> three at 1337 ms from start
    val nums = (1..3).asFlow().onEach { delay(300) } // 发射数字 1..3，间隔 300 毫秒
    val strs = flowOf("one", "two", "three").onEach { delay(400) } // 每 400 毫秒发射一次字符串
    val startTime = System.currentTimeMillis() // 记录开始的时间
    nums.zip(strs) { a, b -> "$a -> $b" } // 使用“zip”组合单个字符串
        .collect { value -> // 收集并打印
            println("$value at ${System.currentTimeMillis() - startTime} ms from start")
        }
}*/

/*
fun requestFlow(i: Int): Flow<String> = flow {
    emit("$i: First")
    delay(500) // 等待 500 毫秒
    emit("$i: Second")
}
*/

/*
fun main() = runBlocking<Unit> {
    (1..3).asFlow().map { requestFlow(it) }
}*/

fun requestFlow(i: Int): Flow<String> = flow {
    emit("$i: First")
    delay(500) // 等待 500 毫秒
    emit("$i: Second")
}

/*
fun main() = runBlocking<Unit> {
    val startTime = System.currentTimeMillis() // 记录开始时间
    (1..3).asFlow().onEach { delay(100) } // 每 100 毫秒发射一个数字
        .flatMapConcat { requestFlow(it) }
        .collect { value -> // 收集并打印
            println("$value at ${System.currentTimeMillis() - startTime} ms from start")
        }
}*/


/*
fun main() = runBlocking<Unit> {
    //1: First at 268 ms from start
    //2: First at 341 ms from start
    //3: First at 443 ms from start
    //1: Second at 769 ms from start
    //2: Second at 842 ms from start
    //3: Second at 945 ms from start
    val startTime = System.currentTimeMillis() // 记录开始时间
    (1..3).asFlow().onEach { delay(100) } // 每 100 毫秒发射一个数字
        .flatMapMerge { requestFlow(it) }
        .collect { value -> // 收集并打印
            println("$value at ${System.currentTimeMillis() - startTime} ms from start")
        }
}*/


/*
fun main() = runBlocking<Unit> {
    //在发出下一个新的流的时候取消先前的流
    //1: First at 196 ms from start
    //2: First at 300 ms from start
    //3: First at 401 ms from start
    //3: Second at 904 ms from start
    val startTime = System.currentTimeMillis() // 记录开始时间
    (1..3).asFlow().onEach { delay(100) } // 每 100 毫秒发射一个数字
        .flatMapLatest { requestFlow(it) }
        .collect { value -> // 收集并打印
            println("$value at ${System.currentTimeMillis() - startTime} ms from start")
        }
}*/


fun simple1(): Flow<Int> = flow {
    for (i in 1..3) {
        println("Emitting $i")
        emit(i) // 发射下一个值
    }
}

/*fun main() = runBlocking<Unit> {
    //异常处理
    //Emitting 1
    //1
    //Emitting 2
    //2
    //Caught java.lang.IllegalStateException: Collected 2
    try {
        simple1().collect { value ->
            println(value)
            check(value <= 1) { "Collected $value" }
        }
    } catch (e: Throwable) {
        println("Caught $e")
    }
}*/

/*
fun simple2(): Flow<String> =
    flow {
        for (i in 1..3) {
            println("Emitting $i")
            emit(i) // 发射下一个值
        }
    }
        .map { value ->
            check(value <= 1) { "Crashed on $value" }
            "string $value"
        }

fun main() = runBlocking<Unit> {
    //有一个异常就发送出去
    //Emitting 1
    //string 1
    //Emitting 2
    //Caught java.lang.IllegalStateException: Crashed on 2
    simple2()
        .catch { e -> emit("Caught $e") } // 发射一个异常
        .collect { value -> println(value) }
}*/

//启动流
// 模仿事件流
fun events(): Flow<Int> = (1..3).asFlow().onEach { delay(200) }

/*
fun main() = runBlocking {
    //1
    //2
    //3
    //onCompletion
    //Done
    events().onEach {
        println(it)
    }.onCompletion { println("onCompletion") }.collect()
    println("Done")
}*/

/*
fun main() = runBlocking<Unit> {
    events()
        .onEach { event -> println("Event: $event") }
        .launchIn(this) // <--- 在单独的协程中执行流
    println("Done")
}*/


fun foo(): Flow<Int> = flow {
    for (i in 1..5) {
        println("Emitting $i")
        emit(i)
    }
}

/*
fun main() = runBlocking<Unit> {
    //1
    //Emitting 2
    //2
    //Emitting 3
    //Exception in thread "main" kotlinx.coroutines.JobCancellationException: BlockingCoroutine was cancelled; job=BlockingCoroutine{Cancelled}@ba8a1dc
    foo().collect { value ->
        if (value == 2) cancel()
        println(value)
    }
}
*/
//直接执行流，直到流被执行完
/*fun main() = runBlocking<Unit> {
    //1
    //2
    //3
    //4
    //5
    //Exception in thread "main" kotlinx.coroutines.JobCancellationException: BlockingCoroutine was cancelled; job=BlockingCoroutine{Cancelled}@78e03bb5
    (1..5).asFlow().collect { value ->
        if (value == 3) cancel()
        println(value)
    }
}*/

//可以看出加入cancellable之后繁忙的流可以取消
/*
fun main() = runBlocking<Unit> {
    //1
    //2
    //3
    //Exception in thread "main" kotlinx.coroutines.JobCancellationException: BlockingCoroutine was cancelled; job=BlockingCoroutine{Cancelled}@593634ad
    (1..5).asFlow().cancellable().collect { value ->
        if (value == 3) cancel()
        println(value)
    }
}*/
